#include "base/base_buffer.h"
#include "base/channel.h"
#include "base/count_down_cond.h"
#include "base/cpuinfo.h"
#include "base/eventloop.h"
#include "base/eventloopthreadpool.h"
#include "base/logthread.h"
#include "base/timer.h"
#include "base/thread.h"
#include "base/threadpool.h"
#include "net/accept.h"
#include "net/buffer.h"
#include "net/connect.h"
#include "net/connectpool.h"

#include <signal.h>
#include <sys/time.h>
#include <sys/syscall.h>
#include <unistd.h>
#include <fcntl.h>

#include "net/tcpserver.h"
namespace net_tools{
// void Sigpipe(int sig){
//   NT_LOG_INFO << "BROKEN PIPE" << NT_LOG_ENDL;
// }
std::atomic<int> Tcpserver::connect_number_;

Tcpserver::Tcpserver()
  :count_down_cond_(new Count_down_cond(-1)),
   eventloop_(new Eventloop(_C::CHANNEL_DEFAULT_SIZE,base::Thread::thread_tid_)),
   accept_(new net::Accept([&]{New_connect_call_back();},eventloop_)), //lambda
   accept_fd(accept_->Get_accept_fd()),
   sockaddr_in_size(sizeof(sockaddr_in)),

   cpu_high_level_callback_([]{;}),
   before_loop_todo_([](Eventloop* eventloop){;}),

   connect_over_callback_([](net::Connect* connect){connect->Print_ip_port("connect");}),
   message_callback_(Default_message),
   close_callback_(Default_close),
   lose_callback_(Default_lose)
{
  connect_number_.store(0);
  signal(SIGPIPE,SIG_IGN);
  // sgiaction版本
  // struct sigaction sa;
  // sa.sa_handler = Sigpipe;
  // sigemptyset(&sa.sa_mask);
  // sa.sa_flags = 0;
  // sigaction(SIGPIPE,&sa,NULL);
};
void Tcpserver::Initserver(int thread_number,int work_thread_number,int mempool_block_number,int mempool_block_size,std::string name){
  net::Connect::threadpool_ = new net_tools::base::Threadpool(work_thread_number);
  net_tools::base::Base_buffer::Singleton_pool(mempool_block_size,mempool_block_number);
  thread_todo = [&](Eventloop* eventloop){Default_thread(eventloop,before_loop_todo_,lose_callback_,count_down_cond_);};
  eventloopthreadpool_ = new Eventloopthreadpool(thread_todo,thread_number);
  eventlooppool_ = eventloopthreadpool_->Get_ptr_looppool();
  thread_number_ = eventloopthreadpool_->Get_thread_number();
  count_down_cond_->Add_count(thread_number_ + 1);
  count_down_cond_->Wait();
  delete count_down_cond_;
  base::Thread::p_thread_name_ = new std::string(name);
};
void Tcpserver::Startserver(int port){
  accept_->Init_listen(port);
  eventloop_->Loop();
};
int Tcpserver::Get_connect_number(){
  int connect_number = 0;
  for(int count = 0;count < thread_number_;++count){
    connect_number += (*eventlooppool_)[count]->connectpool_->Get_connect_number();
  }
  return connect_number;
};
void Tcpserver::Add_timejob(const std::string job_name,func todo,timeval begin_time,timeval time,int count){
  eventloop_->Run_every(std::move(todo),begin_time,time,count,std::move(job_name));
};
void Tcpserver::Add_timejob_after_sec(const std::string job_name,func todo,long int begin_time,long int time,int count){
  timeval now_,sec_;
  gettimeofday(&now_,NULL);
  now_.tv_sec += begin_time;
  sec_.tv_sec = time;
  sec_.tv_usec = 0;
  eventloop_->Run_every(std::move(todo),now_,sec_,count,std::move(job_name));
};
void Tcpserver::Del_timejob(const std::string job_name){
  eventloop_->Deltimejob_every(std::move(job_name));
};
net_tools::base::Timeevent* Tcpserver::Add_once_timejob_at(func func_,timeval begin_time){
  return eventloop_->Run_at(std::move(func_),begin_time);
};
net_tools::base::Timeevent* Tcpserver::Add_once_timejob_after(func func_,timeval begin_time){
  return eventloop_->Run_after(std::move(func_),begin_time);
};
net_tools::base::Timeevent* Tcpserver::Add_once_timejob_after_sec(func func_,long int sec){
  timeval now_;
  gettimeofday(&now_,NULL);
  now_.tv_sec += sec;
  return eventloop_->Run_at(std::move(func_),now_);
};
bool Tcpserver::Del_once_timejob(net_tools::base::Timeevent* timeevent){
  return eventloop_->Deltimejob_once(timeevent);
};

void Tcpserver::New_connect(Eventloop* eventloop,sockaddr_in new_connect_addr,int new_connect_fd){
  timeval now;
  gettimeofday(&now,NULL);
  base::Channel* channel = eventloop->Get_free_channel(new_connect_fd);
  net::Connect* connect = eventloop->connectpool_->Get_Connect(channel);
  connect->Set_address(new_connect_addr);
  connect->connect_time_ = now;
  //connect初始化高效 未完成
  connect->buffer_.Set_able_read_callback(message_callback_);
  connect->buffer_.Set_able_close_callback(close_callback_);
  connect->buffer_.Set_able_user_close_callback(Default_user_close);
  eventloop->connectpool_->Add_heartbeat(connect);
  channel->Enableread(); //Enable recv data
  connect_over_callback_(connect);
  channel->Add_to_epoll();
};
void Tcpserver::New_connect_call_back(){
  struct sockaddr_in accept_addr;
  while (true){
    int new_connect_fd = accept4(accept_fd,reinterpret_cast<sockaddr*>(&accept_addr),&sockaddr_in_size,SOCK_NONBLOCK);
    if(new_connect_fd != -1) [[likely]] { //when no connect,return -1
      // if(connect_number_.load() > _C::MAX_CONNECT_NUMBER) [[unlikely]] {
      //   NT_LOG_WARN << "OVER MAX CONNECT NUMBER" << NT_LOG_ENDL;
      //   close(new_connect_fd);
      //   return;
      // }else if(base::Cpuinfo::cpu_loadavg_ >= _C::HIGH_CPU_LOADAVG){
      //   NT_LOG_WARN << "OVER MAX CONNECT NUMBER" << NT_LOG_ENDL;
      //   close(new_connect_fd);
      //   cpu_high_level_callback_();
      //   return;
      // }
      net::Socket::Set_noblocked(new_connect_fd);
      connect_number_++;
      Eventloop* next_eventloop = eventloopthreadpool_->Get_next_eventloop();
      next_eventloop->Run_in_thread([this,accept_addr,new_connect_fd,next_eventloop]{New_connect(next_eventloop,accept_addr,new_connect_fd);});
    }else{
      return;
    }
  }
};

void Tcpserver::Default_thread(Eventloop* eventloop,eventloop_func before_loop_todo,connect_func lose_callback,Count_down_cond* count_down_cond){
  eventloop->Init_thread_id();
  net::Connectpool* connectpool = new net::Connectpool(eventloop,1024);
  eventloop->connectpool_ = connectpool;
  eventloop->connectpool_->Set_heartbeat(std::move(lose_callback));
  eventloop->connectpool_->Start_heartbeat();
  count_down_cond->Signal(); //初始化完成
  before_loop_todo(eventloop);
  eventloop->Loop();
};

void Tcpserver::Default_message(net::Connect* connect){
  connect->Read_callback();
};

void Tcpserver::Default_close(net::Connect* connect){
  connect->Heartbeat_Close_connect_todo();
  connect->Close_callback();
  connect_number_--;
};
void Tcpserver::Default_lose(net::Connect* connect){
  connect->Lose_heartbeat_callback();
  connect_number_--;
};

void Tcpserver::Default_user_close(net::Connect* connect){
  connect->Heartbeat_Close_connect_todo();
  connect->User_want_close_callback();
  connect_number_--;
};

} // namespace net_tools